package com.shujia.opt

import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo6MapJoin {
  def main(args: Array[String]): Unit = {
    /**
     * 使用coalesce合并小文件
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val studentsDF: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
      .load("spark/data/students.csv")

    val scoresDF: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,subject_id STRING,score INT")
      .load("spark/data/score.txt")

    /**
     * 在local模式下，会对sql进行优化，使用广播变量，思想类似于mapjoin
     *
     * 放到集群中运行，需要手动设置广播的数据
     * 在DSL代码中使用语法糖来使用广播数据
     *
     * 一般来说，大表join小表，小表在DSL函数中使用hint语法将小表广播出去 小表一般指的是小于1G的数据
     *
     * 开启两个作业job
     * 第一个job：将小表数据拉到Driver端，从Driver端广播出去
     * 第二个job：在Executor内部与广播的数据进行数据关联
     *
     */
    val resDF: DataFrame = scoresDF.join(studentsDF.hint("broadcast"), "id")

    resDF.show()

    while (true){

    }

  }

}
